package com.myzhouye.business.sse.server;/***
 *                    _ooOoo_
 *                   o8888888o
 *                   88" . "88
 *                   (| -_- |)
 *                    O\ = /O
 *                ____/`---'\____
 *              .   ' \\| |// `.
 *               / \\||| : |||// \
 *             / _||||| -:- |||||- \
 *               | | \\\ - /// | |
 *             | \_| ''\---/'' | |
 *              \ .-\__ `-` ___/-. /
 *           ___`. .' /--.--\ `. . __
 *        ."" '< `.___\_<|>_/___.' >'"".
 *       | | : `- \`.;`\ _ /`;.`/ - ` : | |
 *         \ \ `-. \_ __\ /__ _/ .-` / /
 * ======`-.____`-.___\_____/___.-`____.-'======
 *                    `=---='
 *
 * .............................................
 *          佛祖保佑             永无BUG
 */

/**
 * @program: maple-main
 * @description:
 * @author: 那条蠢鱼
 * @create: 2024-01-02 14:00
 **/
import com.alibaba.excel.util.CollectionUtils;
import lombok.RequiredArgsConstructor;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.MapUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.cache.Cache;
import org.springframework.data.redis.cache.RedisCache;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.http.MediaType;
import org.springframework.stereotype.Component;
import org.springframework.web.client.RestTemplate;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.IOException;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;
import java.util.stream.Collectors;

import static com.myzhouye.business.sse.server.SseConstants.PUSH_REDIS_STATUS_KEY_FMT;

/**
 * SseServer业务封装类来操作SEE
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class SseServer {

    private final RedisTemplate redisTemplate;

    /**
     * messageId的 SseEmitter对象映射集
     */
    private static Map<String, Map<String, SseEmitter>> sseGroupMap = new ConcurrentHashMap<>();

    /**
     * 创建sse连接
     *
     * @param uid - 消息id（唯一）
     * @return
     */
    public SseEmitter createConnect(String uid) {
        SseEmitter sseEmitter = createConnect(uid, SseConstants.DEFAULT_GROUP);
        return sseEmitter;
    }

    @SneakyThrows
    public SseEmitter createConnect(String uid, String groupId) {
        Map<String, SseEmitter> sseEmitterMap=sseGroupMap.get(groupId);
        if(null==sseEmitterMap){
            sseEmitterMap=new ConcurrentHashMap<>();
            sseGroupMap.put(groupId,sseEmitterMap);
        }
        /**
         * 设置连接超时时间。0表示不过期，默认是30秒，超过时间未完成会抛出异常
         */
//        SseEmitter sseEmitter = new SseEmitter(60*1000L);
        SseEmitter sseEmitter = new SseEmitter(0L);


        // 注册回调
        sseEmitter.onCompletion(completionCallBack(groupId,uid));
        sseEmitter.onTimeout(timeOutCallBack(groupId,uid));
        sseEmitter.onError(errorCallBack(groupId,uid));

        SseEmitter old = sseEmitterMap.get(uid);
        if(null!=old){
            //关闭旧连接
            old.complete();
        }

        sseEmitterMap.put(uid, sseEmitter);

        //记录一下连接总数。数量+1
        log.info("==>新用户接入");
        log.info("==>groupId:{}", groupId);
        log.info("==>uid:{}", uid);
        //推送最近一次的消息
        Object lastMsg =redisTemplate.opsForValue().get(String.format(PUSH_REDIS_STATUS_KEY_FMT,groupId));
        if(null==lastMsg){
            sseEmitter.send("welcome!");
        }else{
            sseEmitter.send(lastMsg);
        }

        return sseEmitter;
    }

    /**
     * 给指定 messageId发消息
     *
     * @param uid - 消息id（唯一）
     * @param message   - 消息文本
     */
    public boolean sendMessage(String groupId,String uid, String message) {
        Map<String, SseEmitter> sseEmitterMap = sseGroupMap.get(groupId);
        if (null!=sseEmitterMap&&sseEmitterMap.containsKey(uid)) {
            try {
                log.debug("({})[{}]--->发送消息:{}", LocalDateTime.now(),uid,message);
                SseEmitter sseEmitter = sseEmitterMap.get(uid);
                sseEmitter.send(message);
                log.debug("({})[{}]--->发送消息完毕",LocalDateTime.now(),uid);
                return true;
            } catch (IOException e) {
                log.error("发送消息异常 ==> messageId={}, 异常信息：", uid, e.getMessage());
                return false;
            }
        } else {
            log.error("连接不存在或者超时， messageId={}", uid);
            return false;
        }
    }

    /**
     * 给所有 messageId广播发送消息
     *
     * @param message
     */
    public void batchAllSendMessage(String groupId,String message) {
        Map<String, SseEmitter> sseEmitterMap = sseGroupMap.get(groupId);
        if (null!=sseEmitterMap){
            sseEmitterMap.forEach((messageId, sseEmitter) -> {
                try {
                    sseEmitter.send(message, MediaType.APPLICATION_JSON);
                    //缓存当前组最后一次的推送状态
                    redisTemplate.opsForValue().set(String.format(PUSH_REDIS_STATUS_KEY_FMT,groupId),message);
                } catch (IOException e) {
                    log.error("广播发送消息异常 ==> messageId={}, 异常信息：", messageId, e.getMessage());
                    removeMessageId(groupId,messageId);
                }
            });
        }
    }

    public void aliveCheck(String groupId) {
        Map<String, SseEmitter> sseEmitterMap = sseGroupMap.get(groupId);
        if (null!=sseEmitterMap){
            for (Map.Entry<String, SseEmitter> entry : sseEmitterMap.entrySet()) {
                String messageId = entry.getKey();
                SseEmitter sseEmitter = entry.getValue();
                try {
                    SseEmitter.SseEventBuilder obj=SseEmitter.event().comment("heart");
                    sseEmitter.send(obj);
                } catch (IOException e) {
                    log.error("广播发送消息异常 ==> messageId={}, 异常信息：", messageId, e.getMessage());
                    removeMessageId(groupId, messageId);
                }
            }
        }
    }

    /**
     * 给指定 messageId集合群发消息
     *
     * @param messageIds
     * @param message
     */
    public void batchSendMessage(String groupId,List<String> messageIds, String message) {
        if (CollectionUtils.isEmpty(messageIds)) {
            return;
        }
        // 去重
        messageIds = messageIds.stream().distinct().collect(Collectors.toList());
        messageIds.forEach(userId -> sendMessage(groupId,userId, message));
    }


    /**
     * 移除 MessageId
     *
     * @param uid
     */
    public void removeMessageId(String groupId,String uid) {
        Map<String, SseEmitter> sseEmitterMap = sseGroupMap.get(groupId);
        if(null==sseEmitterMap){
            return;
        }
        sseEmitterMap.remove(uid);
        //数量-1
        log.info("remove messageId={}", uid);
    }

    /**
     * 获取所有的 MessageId集合
     *
     * @return
     */
    public List<String> getMessageIds(String groupId) {
        Map<String, SseEmitter> sseEmitterMap = sseGroupMap.get(groupId);
        return null!=sseEmitterMap?new ArrayList<>(sseEmitterMap.keySet()):new ArrayList<>();
    }

    /**
     * 获取当前连接总数
     *
     * @return
     */
    public int getConnectTotal(String groupId) {
        Map<String, SseEmitter> sseEmitterMap = sseGroupMap.get(groupId);
        return null!=sseEmitterMap?sseEmitterMap.size():0;
    }

    /**
     * 断开SSE连接时的回调
     *
     * @param uid
     * @return
     */
    private Runnable completionCallBack(String groupId,String uid) {
        return () -> {
            log.info("结束连接 ==>groupId={}， messageId={}",groupId, uid);
//            removeMessageId(groupId,uid);
        };
    }

    /**
     * 连接超时时回调触发
     *
     * @param uid
     * @return
     */
    private Runnable timeOutCallBack(String groupId,String uid) {
        return () -> {
            log.info("连接超时 ==> messageId={}", uid);
            removeMessageId(groupId,uid);
        };
    }

    /**
     * 连接报错时回调触发。
     *
     * @param uid
     * @return
     */
    private Consumer<Throwable> errorCallBack(String groupId,String uid) {
        return throwable -> {
            log.error("连接异常 ==> messageId={}", uid);
            removeMessageId(groupId,uid);
        };
    }
}

